package com.shujia.flink.source

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.source.SourceFunction

object Demo2SourceFunaction {

  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
      *
      * 使用自定义source
      */

    val ds: DataStream[Int] = env.addSource(new MySource)

    ds.print()


    env.execute()


  }
}

/**
  * 自定义source, 时间SourceFunaction接口
  *
  */

class MySource extends SourceFunction[Int] {
  /**
    * run  用于生产数据，把数据发送到下游算子
    * 任务开始的时候运行一次
    *
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0

    while (true) {
      i += 1

      //将数据发送到下游算子
      ctx.collect(i)

      Thread.sleep(1000)
    }

  }

  /**
    * 任务取消的时候会执行的一个方法，  一般用于回收资源
    * 任务取消的时候运行一次
    *
    */

  override def cancel(): Unit = {

  }
}
